<?php
// +----------------------------------------------------------------------
// | ThinkPHP [ WE CAN DO IT JUST THINK IT ]
// +----------------------------------------------------------------------
// | Copyright (c) 2006-2015 http://thinkphp.cn All rights reserved.
// +----------------------------------------------------------------------
// | Licensed ( http://www.apache.org/licenses/LICENSE-2.0 )
// +----------------------------------------------------------------------
// | Author: yunwuxin <448901948@qq.com>
// +----------------------------------------------------------------------
declare(ticks=1);

namespace randy\queue;

use Closure;
use swoole_process;
use Exception;
use think\Cache;
use InvalidArgumentException;

class Listener {
	private $maxProcesses = 30;
	private $child = 0;
	private $masterRedis;

	/**
	 * @var int
	 */
	protected $sleep = 3;

	/**
	 * @var int
	 */
	protected $maxTries = 0;


	/**
	 * @var \Closure|null
	 */
	protected $outputHandler;

	protected $stopStatus = false;

	protected $options = [
		'expire'     => 60,
		'default'    => 'default',
		'host'       => '127.0.0.1',
		'port'       => 6379,
		'password'   => '',
		'select'     => 0,
		'timeout'    => 0,
		'persistent' => false
	];

	/**
	 * Listener constructor.
	 *
	 * @throws Exception
	 */
	public function __construct() {
		$this->masterRedis = $this->redis_client();

		pcntl_signal(SIGCHLD, [
			$this,
			"signal_handler"
		]);
		set_time_limit(0);
		ini_set('default_socket_timeout', -1); //队列处理不超时,解决redis报错:read error on connection
	}

	private function redis_client() {
		if (!extension_loaded('redis')) {
			throw new Exception('redis扩展未安装');
		}
		$options = config('queue');
		if (!empty($options)) {
			$this->options = array_merge($this->options, $options);
		}

		$func = $this->options['persistent'] ? 'pconnect' : 'connect';
		$redis = new \Redis();
		$redis->$func($this->options['host'], $this->options['port'], $this->options['timeout']);

		if ('' != $this->options['password']) {
			$redis->auth($this->options['password']);
		}


		if (0 != $this->options['select']) {
			$redis->select($this->options['select']);
		}
		return $redis;
	}

	/**
	 *
	 * @param $queue
	 * @param $delay
	 * @param $memory
	 * @param $timeout
	 *
	 * @param $tries
	 *
	 * @author : Randy_chen
	 * @Date   : 2020/5/18
	 * @Time   : 22:35
	 */
	public function listen($queue, $delay, $memory, $timeout, $tries) {
		$queue = $this->getQueue($queue);
		while (true) {
			//子进程运行完成且为停止指令时, 释放主进程
			if ($this->checkRestart()) {
				echo '进程停止中........' . PHP_EOL;
				sleep(1);
				continue;
			}


			if ($this->child < $this->maxProcesses) {
				//接收到了停止的指令则不再开启子进程
				$rds = $this->masterRedis;
				$data_pop = $rds->brpop($queue, 3);//无任务时,阻塞等待
				if (!$data_pop) {
					continue;
				}
				$job = $data_pop[1];
				if ($data_pop[1] !== false) {
					$this->masterRedis->zAdd($queue . ':reserved', time() + $this->options['expire'], $job);
				}

				$this->child++;
				$worker = new Worker();
				$process = new swoole_process([
					                              $worker,
					                              'process'
				                              ]);
				$process->name("php xtask:start");
				$data['queue'] = $queue;
				$data['delay'] = $delay;
				$data['memory'] = $memory;
				$data['timeout'] = $timeout;
				$data['tries'] = $tries;
				$data['data'] = $data_pop[1];
				$process->write(json_encode($data));
				$pid = $process->start();
				echo '子进程开启,ID:' . $pid . PHP_EOL;

			} else {
				echo '超过最大进程,等待释放' . PHP_EOL;
				sleep(3);
			}
			if ($this->memoryExceeded($memory)) {
				$this->stop();
			}
		}
	}

	//检测是否需要重启进程
	public function checkRestart() {
		if ($lastRestart = $this->getTimestampOfLastQueueRestart()) {
			$this->queueShouldRestart() && $this->stop() && $this->clearRestartTag();
		}

		//检测停止状态
		if ($this->stopStatus) {
			$this->child == 0 && $this->stop(true);
			return true;
		}
	}

	/**
	 * 获取上次重启守护进程的时间
	 *
	 * @return int|null
	 */
	protected function getTimestampOfLastQueueRestart() {

		return Cache::get('think:queue:restart');
	}

	/**
	 * 清空重启标记
	 *
	 * @return int|null
	 */
	protected function clearRestartTag() {
		return Cache::rm('think:queue:restart');
	}

	/**
	 * 检查是否要重启守护进程
	 *
	 *
	 * @return bool
	 */
	protected function queueShouldRestart() {
		return $this->getTimestampOfLastQueueRestart() < time();
	}

	/**
	 * 进程关闭信号处理
	 *
	 * @param $signal
	 *
	 * @author : Randy_chen
	 * @Date   : 2020/5/19
	 * @Time   : 12:22
	 */
	private function signal_handler($signal) {
		switch ($signal) {
			case SIGCHLD:
				while ($ret = swoole_process::wait(false)) {
					echo "释放的进程PID={$ret['pid']}\n";
					$this->child--;
				}
		}
	}

	/**
	 * 获取队列名
	 *
	 * @param string|null $queue
	 *
	 * @return string
	 */
	protected function getQueue($queue) {
		return 'queues:' . ($queue ?: $this->options['default']);
	}


	/**
	 * Sleep the script for a given number of seconds.
	 *
	 * @param int $seconds
	 *
	 * @return void
	 */
	public function sleep($seconds) {
		sleep($seconds);
	}


	/**
	 * @param int $memoryLimit
	 *
	 * @return bool
	 */
	public function memoryExceeded($memoryLimit) {
		return (memory_get_usage() / 1024 / 1024) >= $memoryLimit;
	}

	/**
	 * @param bool $flag
	 *
	 * @return bool
	 */
	public function stop($flag = false) {
		if ($flag) {
			echo 'exit Success';
			die;
		} else {
			return $this->stopStatus = true;
		}
	}

	/**
	 * 移动延迟任务
	 *
	 * @param string $from
	 * @param string $to
	 * @param bool   $attempt
	 */
	public function migrateExpiredJobs($from, $to, $attempt = true) {
		$this->masterRedis->watch($from);

		$jobs = $this->getExpiredJobs(
			$from, $time = time()
		);
		if (count($jobs) > 0) {
			$this->transaction(function () use ($from, $to, $time, $jobs, $attempt) {
				$this->removeExpiredJobs($from, $time);
				$this->pushExpiredJobsOntoNewQueue($to, $jobs, $attempt);
			});
		}
		$this->masterRedis->unwatch();
	}

	/**
	 * 重新发布到期任务
	 *
	 * @param string  $to
	 * @param array   $jobs
	 * @param boolean $attempt
	 */
	protected function pushExpiredJobsOntoNewQueue($to, $jobs, $attempt = true) {
		if ($attempt) {
			foreach ($jobs as &$job) {
				$attempts = json_decode($job, true)['attempts'];
				$job = $this->setMeta($job, 'attempts', $attempts + 1);
			}
		}
		call_user_func_array([
			                     $this->masterRedis,
			                     'rPush'
		                     ], array_merge([$to], $jobs));
	}

	protected function setMeta($payload, $key, $value) {
		$payload = json_decode($payload, true);
		$payload[$key] = $value;
		$payload = json_encode($payload);

		if (JSON_ERROR_NONE !== json_last_error()) {
			throw new InvalidArgumentException('Unable to create payload: ' . json_last_error_msg());
		}

		return $payload;
	}

	/**
	 * redis事务
	 *
	 * @param \Closure $closure
	 */
	protected function transaction(\Closure $closure) {
		$this->masterRedis->multi();
		try {
			call_user_func($closure);
			if (!$this->masterRedis->exec()) {
				$this->masterRedis->discard();
			}
		} catch (Exception $e) {
			$this->masterRedis->discard();
		}
	}

	/**
	 * 获取所有到期任务
	 *
	 * @param string $from
	 * @param int    $time
	 *
	 * @return array
	 */
	protected function getExpiredJobs($from, $time) {
		return $this->masterRedis->zRangeByScore($from, '-inf', $time);
	}

	/**
	 * 删除过期任务
	 *
	 * @param string $from
	 * @param int    $time
	 *
	 * @return void
	 */
	protected function removeExpiredJobs($from, $time) {
		$this->masterRedis->zRemRangeByScore($from, '-inf', $time);
	}


	/**
	 * @param \Closure $outputHandler
	 *
	 * @return void
	 */
	public function setOutputHandler(Closure $outputHandler) {
		$this->outputHandler = $outputHandler;
	}

	/**
	 * @return int
	 */
	public function getSleep() {
		return $this->sleep;
	}

	/**
	 * @param int $sleep
	 *
	 * @return void
	 */
	public function setSleep($sleep) {
		$this->sleep = $sleep;
	}

	/**
	 * @param int $tries
	 *
	 * @return void
	 */
	public function setMaxTries($tries) {
		$this->maxTries = $tries;
	}
}
